/**
 * Copyright 2007-2008 Alex Tkachman
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.ait.actors;

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;

public class Scheduler {
    private static final Scheduler globalScheduler = new Scheduler();

    final ArrayList<WorkerThread> workers = new ArrayList<WorkerThread> ();

    final CountDownLatch start = new CountDownLatch(1);
    private static DebugThread debugThread;

    private Scheduler () {
        int len = Runtime.getRuntime().availableProcessors() * 2;
        for (int i = 0; i != len; ++i) {
            workers.add(createWorker(i));
        }
        start.countDown();
    }

    private static WorkerThread getWorker() {
        final Thread thread = Thread.currentThread();
        WorkerThread workerThread;
        if (thread instanceof WorkerThread) {
            workerThread = (WorkerThread) thread;
        }
        else {
            workerThread = globalScheduler.getRandomThread();
        }
        return workerThread;
    }

    public static void schedule (Callable runnable) {
        getWorker().schedule (runnable);
    }

    public static void schedule (Callable runnable []) {
        getWorker().schedule (runnable);
    }

    public static <V> V execute (Callable<V> runnable) throws Exception {
        if (runnable instanceof Continuation)
          return execute((Continuation<V>)runnable);
        else
          return execute((Continuation<V>)new Continuation.CallableAdapter(runnable));
    }

    public static <V> V execute (Continuation<V> runnable) throws Exception {
        return runnable.execute();
    }

    public static Object [] execute (final Callable...runnable) throws Exception {
        return new ResultCollectingContinuation(runnable).execute();
    }

    WorkerThread getRandomThread() {
        int index = (int) (Math.random() * workers.size());
        return workers.get(index);
    }

    public static void throwControl() throws ControlException {
        final Thread thread = Thread.currentThread();
        if (thread instanceof WorkerThread)
           ((WorkerThread)thread).throwControl ();
        else
           throw new ControlException ();
    }

    public static void debug (Object s) {
        if (debugThread == null) {
          debugThread = new DebugThread ();
          debugThread.start();
        }

        debugThread.out(s);
    }

    public static class DebugThread extends Thread {
        private LinkedBlockingQueue queue = new LinkedBlockingQueue ();
        public void run() {
            while(true) {
                try {
                    final Msg s = (Msg) queue.take();
                    if (s.msg instanceof Throwable)
                      ((Throwable)s.msg).printStackTrace(System.out);
                    else
                      System.out.println(s.thread.getName() + ": "+ s.msg);
                } catch (InterruptedException e) {
                }
            }
        }

        public void out (Object s) {
            try {
                queue.put(new Msg(s));
            } catch (InterruptedException e) {
            }
        }

        private static class Msg {
            Thread thread = Thread.currentThread();
            Object msg;

            Msg (Object msg) {
                this.msg = msg;
            }
        }
    }

    private static class ResultCollectingContinuation extends Continuation<Object[]> {
        private final Callable[] runnable;

        public ResultCollectingContinuation(Callable[] runnable) {
            this.runnable = runnable;
        }

        protected Object [] doCall() throws Exception {
            if (isNestedCompleted()) {
                return results;
            }

            schedule(runnable);
            return null;
        }
    }

    protected WorkerThread createWorker(int index) {
        return new WorkerThread(this, index);
    }
}
